package com.lyrcsoft.rxjava1x.operators.filter;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Action1;
import rx.functions.Func1;

import java.util.concurrent.TimeUnit;

/**
 * @author fujc2dev@126.com
 * @date 2024-11-21
 */
public class takeUntil {
    /**
     * takeUntil(Observable)订阅并开始发射原始Observable，同时监视我们提供的第二个Observable。
     * 如果第二个Observable发射了一项数据或者发射了一个终止通知，
     * takeUntil()返回的Observable会停止发射原始Observable并终止。
     * <p>
     * takeUntil(Func1)通过Func1中的call方法来判断是否需要终止发射数据。
     *
     * @param args
     */
    public static void main(String[] args) throws InterruptedException {
        // takeUntil可以传Observable也可以传Func1
        Observable.interval(300, TimeUnit.MILLISECONDS)
                .takeUntil(Observable.interval(800, TimeUnit.MILLISECONDS))
                .subscribe(new Subscriber<Long>() {
                    @Override
                    public void onCompleted() {
                        System.out.println("--------------------------------------takeUntil Observable onCompleted--------------------------------------");
                        unsubscribe();
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(Long aLong) {
                        System.out.println(aLong);
                    }
                });


        Thread.sleep(5000);
        System.out.println("--------------------------------------takeUntil Func1 Begin--------------------------------------");
        Observable.just(1, 2, 3, 4, 5, 6, 7)
                .takeUntil(new Func1<Integer, Boolean>() {
                    @Override
                    public Boolean call(Integer integer) {
                        return integer >= 5;
                    }
                }).subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        System.out.println(integer);
                    }
                });
    }
}
